#include <list>
#include <set>
#include <unordered_set>
#include <map>
#include <unordered_map>

#include "../include/ceph_mock/Objecter.h"

#define ceph_assert(e) assert(e)

namespace ceph_mock {
// 基础类型
template<class T>
inline void encode_raw(const T &t, ceph::buffer::list &bl) {
  bl.append((char*)&t, sizeof(t));
}
template<class T>
inline void decode_raw(T &t, ceph::buffer::list &bl) {
  bl.copy((char*)&t, sizeof(t));
}

inline void encode(const uint64_t &size, ceph::buffer::list &bl) {
  encode_raw<uint64_t>(size, bl);
}
inline void decode(uint64_t &size, ceph::buffer::list &bl) {
  decode_raw<uint64_t>(size, bl);
}
inline void encode(const double &mtime, ceph::buffer::list &bl) {
  encode_raw<double>(mtime, bl);
}
inline void decode(double &mtime, ceph::buffer::list &bl) {
  decode_raw<double>(mtime, bl);
}

// string : [len:string]
inline void encode(const std::string &s, ceph::buffer::list &bl) {
  uint32_t len = s.length();
  encode_raw<uint32_t>(len, bl);
  bl.append(s.data(), len);
}
inline void decode(std::string &s, ceph::buffer::list &bl) {
  uint32_t pos = 0;

  uint32_t len;
  bl.copy((char*)&len, sizeof(len), pos);
  pos += sizeof(len);
  
  s.clear();
  bl.copy(const_cast<char*>(s.data()), len, pos);
}

// set : {n:[len:string]...}
inline void encode(const std::set<std::string> &keys, ceph::buffer::list &bl) {
  uint32_t n = keys.size();
  encode_raw<uint32_t>(n, bl);
  for(auto k : keys) {
    encode(k, bl);
  }
}
inline void decode(std::set<std::string> &keys, ceph::buffer::list &bl) {
  uint32_t pos = 0;
  
  uint32_t n;
  bl.copy((char*)&n, sizeof(n), pos);
  pos += sizeof(n);
  
  for(int i = 0; i < n; ++i) {
    uint32_t len;
    bl.copy((char*)&len, sizeof(len), pos);
    pos += sizeof(len);
    
    std::string s;
    bl.copy(const_cast<char*>(s.data()), len, pos);
    pos += len;
    
    keys.insert(s);
  }
}
inline void encode(const std::unordered_set<std::string> &keys, ceph::buffer::list &bl) {
  uint32_t n = keys.size();
  encode_raw<uint32_t>(n, bl);
  for(auto k : keys) {
    encode(k, bl);
  }
}
inline void decode(std::unordered_set<std::string> &keys, ceph::buffer::list &bl) {
  uint32_t pos = 0;
  
  uint32_t n;
  bl.copy((char*)&n, sizeof(n), pos);
  pos += sizeof(n);
  
  for(int i = 0; i < n; ++i) {
    uint32_t len;
    bl.copy((char*)&len, sizeof(len), pos);
    pos += sizeof(len);
    
    std::string s;
    bl.copy(const_cast<char*>(s.data()), len, pos);
    pos += len;
    
    keys.insert(s);
  }
}

// map : {n:[len:first:len:second]...}
inline void encode(const std::map<std::string, ceph::buffer::list> &keys, ceph::buffer::list &bl) {
  uint32_t n = keys.size();
  encode_raw<uint32_t>(n, bl);
  for(auto k : keys) {
    encode(k.first, bl);
    encode(k.second, bl);
  }
}
inline void decode(std::map<std::string, ceph::buffer::list> &keys, ceph::buffer::list &bl) {
  uint32_t pos = 0;
  
  uint32_t n;
  bl.copy((char*)&n, sizeof(n), pos);
  pos += sizeof(n);
  
  for(int i = 0; i < n; ++i) {
    uint32_t len;
    bl.copy((char*)&len, sizeof(len), pos);
    pos += sizeof(len);
    
    std::string s1;
    bl.copy(const_cast<char*>(s1.data()), len, pos);
    pos += len;
    
    bl.copy((char*)&len, sizeof(len), pos);
    pos += sizeof(len);

    std::string s2;
    bl.copy(const_cast<char*>(s2.data()), len, pos);
    pos += len;
    
    keys[s1] = s2;
  }  
}

// TODO (yanyi) Objecter::_maybe_do_op_on_pmem()
/**
 * MDS PMem 修改，将远端 PrimaryLogPG 上的工作搬到 MDS 本地在主备 PMem 空间上直接执行
 * @note 本函数是（Objecter::_op_submit_with_budget() 中）Objecter::_op_submit() 的平行替代
 * @note 目前实现不涉及并发读写 Objecter 状态，`sul` 这个锁可有可无，但保险起见还是加上了
 * @param[in,out] op Objecter::_op_submit() 的操作事务输入， Objecter::handler_osd_op_reply()
 *    接收的事务操作结果输出，复用同一个 `OSDOp` 向量上的 `indata` 与 `outdata` 空间
 * @param[in,out] sul Objecter 状态锁。目前只是做个断言，没有涉及到修改这个锁的状态
 * @param[out] result 原先封装在 `MOSDOpReply` 中的整个事务的操作返回值
 * @return 该次调用 __（不包括其回调）__ 是否在 MDS 本地完成
 */
bool Objecter::_maybe_do_op_on_pmem(
      Op *op, std::shared_lock<std::shared_mutex> &sul,
      int &result)
{
  /* 检查我们是否在 MDS 上 */
  if (!this->rpmkv)
    return false;

  using namespace std::literals;
  ceph_assert(op);

  /* 在 Objecter::op_submit() 中调用需要持有 Objecter 对象上的读锁 */
  ceph_assert(sul);
  /* 更新 perf-cnt，尽量做到与原 _op_submit() 逻辑兼容 */
  // _send_op_account(op);

  /* NOTE: 不需要 OSDSessoin，因为我们并不与 OSD 通信。 */

  const std::string &oid = op->target.base_oid.name;
  const auto mtime = op->mtime;

  {
    ///* 这里直接加个 KV 上的写锁，不然需要在事务里面再管理锁级别，容易出问题 */
    //std::scoped_lock _l(rpmkv->lock);
    /* 根据事务内容对 KV 上读写锁 */
    std::unique_lock<std::shared_mutex> _xl;
    std::shared_lock<std::shared_mutex> _rl;
    if constexpr (true) {
      bool may_write = false;
      for (const auto &osd_op : op->ops) {
        if (may_write)
          break;
        switch (osd_op.op.op) {
        case CEPH_OSD_OP_CREATE:
        case CEPH_OSD_OP_DELETE:
        case CEPH_OSD_OP_ZERO:
        case CEPH_OSD_OP_WRITE:
        case CEPH_OSD_OP_WRITEFULL:
        case CEPH_OSD_OP_OMAPSETHEADER:
        case CEPH_OSD_OP_OMAPSETVALS:
        case CEPH_OSD_OP_OMAPRMKEYS:
        case CEPH_OSD_OP_OMAPCLEAR:
        case CEPH_OSD_OP_SETXATTR:
        case CEPH_OSD_OP_RMXATTR:
        case CEPH_OSD_OP_TRIMTRUNC:
        case CEPH_OSD_OP_TRUNCATE:
          may_write = true;
        }
      }

      auto &mtx = rpmkv->get_hoid_segment_mutex(oid);
      if (may_write)
        _xl = std::unique_lock(mtx);
      else
        _rl = std::shared_lock(mtx);
    }

    try {
      pmem::obj::transaction::run(rpmkv->get_pool(), [&] {
        constexpr bool enable_zero_copy = true;

        const auto may_create_obj = [&] (const std::string &oid, size_t alloc_size_hint=4<<10)
        {
          auto it = rpmkv->find(oid);
          if (!it)
            it = rpmkv->create(oid, alloc_size_hint);
          return it;
        };
        const auto may_create_obj_nothrow = [&] (
            const std::string &oid, size_t alloc_size_hint=4<<10)
            -> std::pair<decltype(rpmkv->end()), int>
        {
          try {
            auto it = may_create_obj(oid, alloc_size_hint);
            return {it, 0};
          }
          catch (const std::length_error &ex) {
            return {rpmkv->end(), -ENAMETOOLONG};
          }
          catch (const std::out_of_range &ex) {
            return {rpmkv->end(), -EFBIG};
          }
          throw std::runtime_error("unreachable");
        };
        /**
         * （默认创建并）重设对象内容
         * @return
         *    * 0 - 成功
         *    * -ENAMETOOLONG - 对象 oid 太长
         *    * -EFBIG - 空间分配失败
         */
        const auto set_full_obj = [&] (
            const std::string &oid, const ceph::buffer::list &bl,
            size_t alloc_size_hint=4ul<<10) /*noexcept*/ -> int
        {
          const size_t len = bl.length();

          alloc_size_hint = len < alloc_size_hint ?
              alloc_size_hint : std::max(len, 4ul << 20);
          auto it = rpmkv->end();
          try {
            it = may_create_obj(oid, alloc_size_hint);
            /* PERF: 暂时不考虑合并快照优化 */
            it->resize(len);
            if constexpr (enable_zero_copy) {
              char *buf = it->range(0, len);
              ceph_assert(buf);
              // bl.cbegin().copy(len, buf);
              bl.copy(buf, len);

              //rpmkv->get_pool().persist(buf, len);
            }
            else {
              auto buf = std::make_unique<char[]>(len);
              // bl.cbegin().copy(len, buf.get());
              bl.copy(buf.get(), len);
              
              int r = it->write(0, len, buf.get());
              if (r < 0)
                return r;
              if (r < len)
                return -EIO;
            }
          }
          catch (const std::length_error &ex) {
            return -ENAMETOOLONG;
          }
          catch (const std::out_of_range &ex) {
            return -EFBIG;
          }
          return 0;
        };

        /**** OMAP 相关 ****/

        const auto get_omap_header_oid = [] (const std::string &oid)
        { return oid + "##header"; };
        const auto get_omap_kv_oid = [] (const std::string &oid, const std::string &key)
        { return oid + "#" + key; };

        /**
         * 为了支持 OMAP 键上的范围查询，我们将所有键存放在一个独立的对象 "<oid>##keys" 中。
         * 我们允许对象 "<oid>" 存在而与其对应的 "<oid>##keys" 不存在，此时说明该对象没有
         * 使用 OMAP 或其 OMAP 为空。
         */

        const auto get_omap_all_keys_oid = [] (const std::string &oid)
        { return oid + "##keys"; };

        /**
         * 重设某个对象 OMAP 中的所有键
         *
         * @note 本函数默认对象 "<oid>" 是存在的，即创建该对象的 OMAP 是合法的
         *
         * 为了实现方便，"<oid>##keys" 存放的即为使用 Ceph 序列化/反序列化工具封装的
         * std::set<std::string>。且 std::set 本身也支持范围查询算法。
         */
        const auto reset_all_omap_keys = [&] (
            const std::string &oid, const std::set<std::string> &keys) -> int
        {
          ceph::buffer::list bl;
          encode(keys, bl);

          return set_full_obj(get_omap_all_keys_oid(oid), bl, 4ul << 20);
        };
        /**
         * 获取某个对象的 OMAP 中的所有键
         *
         * @todo PERF 考虑缓存常用对象的 OMAP 键表？
         */
        const auto get_all_omap_keys = [&] (const std::string &oid)
            -> std::set<std::string>
        {
          auto it = rpmkv->cfind(get_omap_all_keys_oid(oid));
          if (it == rpmkv->cend())
            return {};
          /* 非空情况 */
          std::set<std::string> keys;
          {
            ceph::buffer::list bl;
            bl.append(it->data(), it->size());
            decode(keys, bl);
          }
          return keys;
        };

        /**** 扩展属性 xattr 相关 ****/

        /**
         * 扩展属性 xattr 的物理实现与 OMAP 类似，只不过 xattr 不要求键有序，仅支持点查询
         */

        const auto get_xattr_oid = [] (const std::string &oid, const std::string &key)
        { return oid + "@" + key; };
        const auto get_xattr_all_keys_oid = [] (const std::string &oid)
        { return oid + "@@keys"; };

        const auto reset_all_xattr_keys = [&] (
            const std::string &oid, const std::unordered_set<std::string> &keys) -> int
        {
          ceph::buffer::list bl;
          encode(keys, bl);
          return set_full_obj(get_xattr_all_keys_oid(oid), bl, 4ul << 20);
        };
        /**
         * @todo PERF 同样地，考虑本地 DRAM 中缓存对象 xattr 键表
         */
        const auto get_all_xattr_keys = [&] (const std::string &oid)
            -> std::unordered_set<std::string>
        {
          auto it = rpmkv->cfind(get_xattr_all_keys_oid(oid));
          if (it == rpmkv->cend())
            return {};
          /* 非空情况 */
          std::unordered_set<std::string> keys;
          {
            ceph::buffer::list bl;
            bl.append(it->data(), it->size());
            decode(keys, bl);
          }
          return keys;
        };

        /**** 实际执行事务 ****/

        result = 0;

        for (auto &osd_op : op->ops) {
          // ldout(cct, 20) << __func__ << " in transaction, performing OSDOp "
			    // << osd_op << dendl;

          osd_op.rval = 0;
          osd_op.outdata.clear();

          /* 规范 truncate_seq/size 参数 */
          if (ceph_osd_op_uses_extent(osd_op.op.op) && osd_op.op.extent.truncate_size == -1ull) {
            osd_op.op.extent.truncate_seq = 0;
            osd_op.op.extent.truncate_size = 0;
          }

          switch (osd_op.op.op) {
          case CEPH_OSD_OP_STAT: {
            auto it = rpmkv->cfind(oid);
            if (it == rpmkv->cend()) {
              osd_op.rval = -ENOENT;
              break;
            }
            // encode(it->size(), osd_op.outdata);
            // encode(ceph::real_clock::from_double(it->mtime()), osd_op.outdata);
            encode(it->size(), osd_op.outdata);
            encode(it->mtime(), osd_op.outdata);
          }
          break;

          case CEPH_OSD_OP_CREATE: {
            auto it = rpmkv->find(oid);
            if (it && (osd_op.op.flags & CEPH_OSD_OP_FLAG_EXCL)) {
              osd_op.rval = -EEXIST;
              break;
            }
            if (!it) {
              try {
                it = rpmkv->create(oid, 4 << 20);
              }
              catch (const std::length_error &ex) {
                osd_op.rval = -ENAMETOOLONG;
                break;
              }
              catch (const std::out_of_range &ex) {
                /* NOTE: OSD 上创建对象并不涉及存储空间的分配，因此总是成功（除非
                    new ObjectState 失败）。但我们这里需要在 PMem 上立即分配空间，
                    因此需要补充这一种错误状态 */
                osd_op.rval = -EFBIG;
                break;
              }
            }
            ceph_assert(it);
            /* 无论 KV 项是否是新创建的，都更新其修改时间 */
            it->touch(mtime);
          }
          break;

          case CEPH_OSD_OP_DELETE: {
            auto blobit = rpmkv->cfind(oid);
            if (!blobit) {
              osd_op.rval = -ENOENT;
              break;
            }
            /* 删除对象 blob 数据 */
            rpmkv->remove(blobit);
            /**
             * 删除对象时应该清空 xattr 与 OMAP。这是必要的，不仅仅出于空间利用率考虑，而是
             * 出于数据操作正确性考虑！存在于上一个同名对象生命周期中的附带值不应该影响发往
             * 下一个对象的请求的处理结果。
             */
            /* 清除对象 xattr */
            {
              /* 删除 xattr 数据 */
              {
                auto keys = get_all_xattr_keys(oid);
                for (const auto &k : keys)
                  rpmkv->remove(get_xattr_oid(oid, k));
              }
              /* 删除 xattr 索引 */
              rpmkv->remove(get_xattr_all_keys_oid(oid));
            }
            /* 清除对象 OMAP */
            {
              /* 删除 OMAP 头段 */
              rpmkv->remove(get_omap_header_oid(oid));
              /* 删除 OMAP KV 数据 */
              {
                auto keys = get_all_omap_keys(oid);
                for (const auto &k : keys)
                  rpmkv->remove(get_omap_kv_oid(oid, k));
              }
              /* 删除 OMAP 索引 */
              rpmkv->remove(get_omap_all_keys_oid(oid));
            }
          }
          break;

          case CEPH_OSD_OP_READ: {
            /* 规范读大小 */
            if (osd_op.op.extent.truncate_seq) {
              /* NOTE: 从 PrimaryLogPG 的源码来看，READ 中的 truncate 并不会修改对象
                  大小（既没有修改 object_info_t，也没有传给 PGBackend），而仅仅是作用于读范围上
                  不知道为什么要这样设计调用参数，但他就是这样写的 */
              osd_op.op.extent.length = std::min(
                  (uint64_t)osd_op.op.extent.length, (uint64_t)(osd_op.op.extent.truncate_size - osd_op.op.extent.offset));
            }
            auto it = rpmkv->cfind(oid);
            /* 请求的 KV 不存在 */
            if (!it) {
              osd_op.rval = -ENOENT;
              break;
            }
            /* 完全越界，不读出任何值，但返回成功 */
            if (osd_op.op.extent.offset >= it->size()) {
              osd_op.op.extent.length = 0;
              osd_op.rval = 0;
              break;
            }

            /* 若传入读大小为 0，则读整个对象（严格来说是读到对象尾，因为 offset 可能非零） */
            if (osd_op.op.extent.length == 0)
              osd_op.op.extent.length = it->size();

            if constexpr (enable_zero_copy) {
              osd_op.op.extent.length = std::min((uint64_t)osd_op.op.extent.length, (uint64_t)(it->size() - osd_op.op.extent.offset));
              osd_op.outdata.append(it->data() + osd_op.op.extent.offset, osd_op.op.extent.length);
            }
            else {
              auto buf = std::make_unique<char[]>(osd_op.op.extent.length);
              int r = it->read(osd_op.op.extent.offset, osd_op.op.extent.length, buf.get());
              if (r >= 0) {
                osd_op.outdata.append(buf.get(), r);
                osd_op.op.extent.length = r;
              }
              else {
                osd_op.op.extent.length = 0;
                /* 暴露底层读接口错误值 */
                osd_op.rval = r;
              }
            }
          }
          break;

          case CEPH_OSD_OP_ZERO: {
            auto it = rpmkv->find(oid);
            /* 如果对象不存在，不做任何事情，且返回值为 0（即使 READ 会失败） */
            if (!it)
              break;
            int r = 0;
            try {
              r = it->zero(osd_op.op.extent.offset, osd_op.op.extent.length);
            }
            catch (const std::out_of_range &ex) {
              osd_op.rval = -EFBIG;
              break;
            }
            if (r < 0) {
              osd_op.rval = r;
              break;
            }
            if (r < osd_op.op.extent.length) {
              osd_op.rval = -EIO;
              break;
            }
            it->touch(mtime);
          }
          break;

          case CEPH_OSD_OP_WRITE: {
            osd_op.indata.assign(4ul<<10, 0);
            osd_op.op.extent.length = 4ul<<10;

            if (osd_op.op.extent.length != osd_op.indata.length()) {
              osd_op.rval = -EINVAL;
              break;
            }
            /* 若不存在，则默认创建对象 */
            auto it = rpmkv->end();
            std::tie(it, osd_op.rval) = may_create_obj_nothrow(oid,
                std::max((uint64_t)(osd_op.op.extent.offset + osd_op.op.extent.length), (uint64_t)(4ul << 20)));
            if (osd_op.rval)
              break;
            ceph_assert(it);
            /* NOTE: 写对齐检查我们就不做了，我们的底层数据结构没有这种限制 */
            /* 先修剪对象长度 */
            /* NOTE: 由于我们所有的操作都是立即同步更新的，因此 seq 的具体值在本实现中并不关心。
                需要注意的是当请求并不是严格按照 seq 到达时，这会导致最终读写结果不正确。但
                写时 trunc 并不常用，至少在之前的源码整理过程中没有发现 MDS 相关逻辑使用，
                更没有上面说到的这种排序的并发使用场景（说到底也不应该有这种场景==），因此就
                只是提供了最简单的 one-shot 支持
               PERF: 此外，基于同样的原因，这里不对修剪写做快照合并优化
               NOTE: 尽管这里的修剪可能将对象裁剪到不包含我们要写的区间，但后续写是可以重新
                把对象填充到能够容纳写区间的大小的。 */
            if (osd_op.op.extent.truncate_seq)
              it->resize(osd_op.op.extent.truncate_size);

            const size_t old_size = it->size();
            if constexpr (enable_zero_copy) {
              char *val = nullptr;
              try {
                val = it->range(osd_op.op.extent.offset, osd_op.op.extent.length);
              }
              catch (const std::out_of_range &ex) {
                osd_op.rval = -EFBIG;
                break;
              }
              ceph_assert(val);
              /* 如果生成了空洞，则需要填零 */
              if (osd_op.op.extent.offset > old_size) {
                /* PERF: 注意这里空洞区域和实际写数据区域是分开快照的，暂时不考虑合并优化 */
                it->zero(old_size, osd_op.op.extent.offset - old_size);
              }
              // indata.cbegin().copy(osd_op.op.extent.length, val);
              osd_op.indata.copy(val, osd_op.op.extent.length);
              //rpmkv->get_pool().persist(val, osd_op.op.extent.length);
            }
            else {
              auto buf = std::make_unique<char[]>(osd_op.op.extent.length);
              // osd_op.indata.cbegin().copy(osd_op.op.extent.length, buf.get());
              osd_op.indata.copy(buf.get(), osd_op.op.extent.length);
              try {
                int r = it->write(osd_op.op.extent.offset, osd_op.op.extent.length, buf.get());
                osd_op.rval = r >= 0 ? 0 : r;
              }
              catch (const std::out_of_range &ex) {
                osd_op.rval = -EFBIG;
              }
              if (osd_op.rval)
                break;
            }
            it->touch(mtime);
          }
          break;

          case CEPH_OSD_OP_WRITEFULL: {
            osd_op.indata.assign(4ul<<20, 0);
            osd_op.op.extent.length = 4ul<<20;

            if (osd_op.op.extent.length != osd_op.indata.length()) {
              osd_op.rval = -EINVAL;
              break;
            }

            osd_op.rval = set_full_obj(oid, osd_op.indata, 4ul << 20);
            if (osd_op.rval)
              break;
            {
              /* PERF: 为了重用 set_full_obj() 代码导致进行了两次索引查询。
                  可以在本实现中展开，或者在 PMemDevice 中设计迭代器缓存 */
              auto it = rpmkv->find(oid);
              ceph_assert(it);
              it->touch(mtime);
            }
          }
          break;

          case CEPH_OSD_OP_OMAPGETHEADER: {
            if (!rpmkv->cfind(oid)) {
              osd_op.rval = -ENOENT;
              break;
            }
            /**
             * 与存放对象 OMAP 所有键的对象 "<oid>##keys" 类似，我们也允许对象 OMAP 头段
             * "<oid>##header" 不存在，此时则意味着 OMAP 头段内容为空。
             */
            auto it = rpmkv->cfind(get_omap_header_oid(oid));
            if (!it)
              break;
            if constexpr (enable_zero_copy) {
              osd_op.outdata.append(it->data(), it->size());
            }
            else {
              auto buf = std::make_unique<char[]>(it->size());
              int r = it->read(0, 0, buf.get());
              if (r < 0)
                osd_op.rval = r;
              else if (r < it->size())
                osd_op.rval = -EIO;
            }
          }
          break;

          case CEPH_OSD_OP_OMAPSETHEADER: {
            osd_op.indata.assign(4ul<<10, 0);
            osd_op.op.extent.length = 4ul<<10;

            if (osd_op.op.extent.length != osd_op.indata.length()) {
              osd_op.rval = -EINVAL;
              break;
            }
            /* 确保对象 blob 存在（我们通过对象 blob 验证对象存在性），顺便更新修改时间 */
            {
              auto blobit = rpmkv->end();
              std::tie(blobit, osd_op.rval) = may_create_obj_nothrow(oid);
              if (osd_op.rval)
                break;
              blobit->touch(mtime);
            }

            osd_op.rval = set_full_obj(get_omap_header_oid(oid), osd_op.indata);
          }
          break;

          case CEPH_OSD_OP_OMAPSETVALS: {
            if (osd_op.op.extent.length != osd_op.indata.length()) {
              osd_op.rval = -EINVAL;
              break;
            }

            // TODO: 构造map
	          std::map<std::string, ceph::buffer::list> to_insert;
            for(int i = 0; i < 10; ++i) {
              std::string key = "key " + to_string(i);
              ceph::buffer::list val(4096, 0);
              to_insert[key] = val;
            }
            // try {
	          //   ceph::buffer::list::const_iterator pt = osd_op.indata.begin();
            //   decode(to_insert, pt);
            // }
            // catch (const ceph::buffer::error &ex) {
            //   osd_op.rval = -EINVAL;
            //   break;
            // }

            /* 确保对象存在，并更新修改时间 */
            {
              auto blobit = rpmkv->end();
              std::tie(blobit, osd_op.rval) = may_create_obj_nothrow(oid);
              if (osd_op.rval)
                break;
              blobit->touch(mtime);
            }

            auto all_keys = get_all_omap_keys(oid);
            for (const auto &kv : to_insert) {
              if (osd_op.rval)
                break;
              all_keys.insert(kv.first);
              osd_op.rval = set_full_obj(get_omap_kv_oid(oid, kv.first), kv.second);
            }
            if (osd_op.rval)
              break;
            osd_op.rval = reset_all_omap_keys(oid, all_keys);
          }
          break;

          case CEPH_OSD_OP_OMAPRMKEYS: {
            {
              auto blobit = rpmkv->find(oid);
              if (!blobit) {
                osd_op.rval = -ENOENT;
                break;
              }
              blobit->touch(mtime);
            }
	          // std::set<std::string> to_remove;

            // try {
	          //   ceph::buffer::list::const_iterator pt = osd_op.indata.begin();
            //   decode(to_remove, pt);
            // }
            // catch (const ceph::buffer::error &ex) {
            //   osd_op.rval = -EINVAL;
            //   break;
            // }

            // 删除所有keys
            auto all_keys = get_all_omap_keys(oid);
            for (const auto &k : all_keys) {
              // all_keys.erase(k);
              rpmkv->remove(get_omap_kv_oid(oid, k));
            }
            osd_op.rval = reset_all_omap_keys(oid, all_keys);
          }
          break;

          case CEPH_OSD_OP_OMAPGETVALS: {
            if (osd_op.op.extent.length != osd_op.indata.length()) {
              osd_op.rval = -EINVAL;
              break;
            }
            if (!rpmkv->cfind(oid)) {
              osd_op.rval = -ENOENT;
              break;
            }
            // std::string start_after;
            // uint64_t max_return;
            // std::string filter_prefix;
            // try {
            //   auto pbl = osd_op.indata.cbegin();
            //   decode(start_after, pbl);
            //   decode(max_return, pbl);
            //   decode(filter_prefix, pbl);
            // }
            // catch (const ceph::buffer::error &ex) {
            //   osd_op.rval = -EINVAL;
            //   break;
            // }
            /* 获取 OMAP KV 数据 */
            /* PERF 直接写到 osd_op.outdata 上 */
            std::map<std::string, ceph::buffer::list> out;
            const auto all_keys = get_all_omap_keys(oid);
            // auto kit = filter_prefix <= start_after ?
            //   all_keys.upper_bound(start_after) : all_keys.lower_bound(filter_prefix);
            // bool truncated = false;
            // for (size_t num = 0;
            //       (kit != all_keys.end()) && kit->substr(0, filter_prefix.length()) == filter_prefix;
            //       kit++, num++) {
            //   if (num >= max_return) {
            //     truncated = true;
            //     break;
            //   }
            //   const auto &k = *kit;
            //   auto vit = rpmkv->cfind(get_omap_kv_oid(oid, k));
            //   ceph_assert(vit);
            //   out[k].append(vit->data(), vit->size());
            // }
            /* 将获取到的结果填充到输出域中 */

            // 读取所有vals
            for (const auto &k : all_keys) {
              const auto it = rpmkv->cfind(get_omap_kv_oid(oid, k));
              if (!it)
                continue;
              out[k].append(it->data(), it->size());
            }
            encode(out, osd_op.outdata);
            /* 是否为不完整读 */
            // encode(truncated, osd_op.outdata);
          }
          break;

          case CEPH_OSD_OP_OMAPGETVALSBYKEYS: {
            if (osd_op.op.extent.length != osd_op.indata.length()) {
              osd_op.rval = -EINVAL;
              break;
            }
            if (!rpmkv->cfind(oid)) {
              osd_op.rval = -ENOENT;
              break;
            }
            // std::set<std::string> to_get;
            // try {
	          //   ceph::buffer::list::const_iterator pt = osd_op.indata.begin();
            //   decode(to_get, pt);
            // }
            // catch (const ceph::buffer::error &ex) {
            //   osd_op.rval = -EINVAL;
            //   break;
            // }
            /* PERF 直接写到 osd_op.outdata 上 */
            // 
            // 读取所有vals
            std::map<std::string, ceph::buffer::list> out;
            const auto all_keys = get_all_omap_keys(oid);
            for (const auto &k : all_keys) {
              const auto it = rpmkv->cfind(get_omap_kv_oid(oid, k));
              if (!it)
                continue;
              out[k].append(it->data(), it->size());
            }
            encode(out, osd_op.outdata);
          }
          break;

          case CEPH_OSD_OP_OMAPCLEAR: {
            {
              auto blobit = rpmkv->find(oid);
              if (!blobit) {
                osd_op.rval = -ENOENT;
                break;
              }
              blobit->touch(mtime);
            }
            const auto all_keys = get_all_omap_keys(oid);
            for (const auto &k : all_keys)
              rpmkv->remove(get_omap_kv_oid(oid, k));
            osd_op.rval = reset_all_omap_keys(oid, {});
          }
          break;

          case CEPH_OSD_OP_SETXATTR: {
            {
              auto blobit = rpmkv->end();
              std::tie(blobit, osd_op.rval) = may_create_obj_nothrow(oid);
              if (osd_op.rval)
                break;
              blobit->touch(mtime);
            }

            // aname和aval赋值
            std::string aname = "layout";
            ceph::buffer::list aval;
            aval.assign(4096, 0);

            // try {
            //   auto bp = osd_op.indata;
            //   bp.copy(osd_op.op.xattr.name_len, aname);
            //   bp.copy(osd_op.op.xattr.value_len, aval);
            // }
            // catch (const ceph::buffer::error &ex) {
            //   osd_op.rval = -EINVAL;
            //   break;
            // }
            auto all_xattrs = get_all_xattr_keys(oid);
            all_xattrs.insert(aname);
            osd_op.rval = set_full_obj(get_xattr_oid(oid, aname), aval);
            if (osd_op.rval)
              break;
            osd_op.rval = reset_all_xattr_keys(oid, all_xattrs);
          }
          break;

          case CEPH_OSD_OP_RMXATTR: {
            {
              auto blobit = rpmkv->find(oid);
              if (!blobit) {
                osd_op.rval = -ENOENT;
                break;
              }
              blobit->touch(mtime);
            }

            // aname赋值
            std::string aname = "layout";

            // try {
            //   auto bp = osd_op.indata.cbegin();
            //   bp.copy(osd_op.op.xattr.name_len, aname);
            // }
            // catch (const ceph::buffer::error &ex) {
            //   osd_op.rval = -EINVAL;
            //   break;
            // }
            auto all_xattrs = get_all_xattr_keys(oid);
            all_xattrs.erase(aname);
            rpmkv->remove(get_xattr_oid(oid, aname));
            osd_op.rval = reset_all_xattr_keys(oid, all_xattrs);
          }
          break;

          case CEPH_OSD_OP_GETXATTR: {
            if (!rpmkv->cfind(oid)) {
              osd_op.rval = -ENOENT;
              break;
            }

            // aname赋值
            std::string aname = "layout";

            // try {
            //   osd_op.indata.cbegin().copy(osd_op.op.xattr.name_len, aname);
            // }
            // catch (const ceph::buffer::error &ex) {
            //   osd_op.rval = -EINVAL;
            //   break;
            // }
            {
              auto it = rpmkv->cfind(get_xattr_oid(oid, aname));
              if (!it) {
                osd_op.rval = -ENODATA;
                break;
              }
              osd_op.op.xattr.value_len = it->size();
              osd_op.outdata.append(it->data(), osd_op.op.xattr.value_len);
            }
          }
          break;

          case CEPH_OSD_OP_TRIMTRUNC:
            osd_op.op.extent.offset = osd_op.op.extent.truncate_size;
            #if __has_cpp_attribute(fallthrough)
            [[fallthrough]];
            #endif
          case CEPH_OSD_OP_TRUNCATE: {
            auto it = rpmkv->end();
            std::tie(it, osd_op.rval) = may_create_obj_nothrow(oid, osd_op.op.extent.offset);
            if (osd_op.rval)
              break;
            ceph_assert(it);
            it->resize(osd_op.op.extent.offset);
            it->touch(mtime);
          }
          break;

          default: {
            // ostringstream msg;
            // msg << "unimplemented op: " << osd_op;
            // throw std::logic_error(msg.str());
          }
          } /* switch (osd_op.op.op) */

          /* 失败直接终止，并回滚（除非 FAILOK） */
          if (osd_op.rval < 0
              && !((osd_op.op.flags & CEPH_OSD_OP_FLAG_FAILOK)
                    && !(osd_op.rval == -EAGAIN) && !(osd_op.rval == -EINPROGRESS))) {
            result = osd_op.rval;
            throw std::runtime_error("aborted with "s + std::strerror(-osd_op.rval));
          }
        } /* for each OSDOp in Op */
      });
    }
    catch (const std::runtime_error &ex) {
      /* 事务执行失败，终止并回滚 */
      // ldout(cct, 10) << __func__ << " rolled-back: " << ex.what() << dendl;
    }
    catch (const std::logic_error &ex) {
      /* 有未实现的逻辑，写到 log 里方便排查 */
      // ldout(cct, 0) << __func__ << " throwing: " << ex.what() << dendl;
      throw ex;
    }
  } /* rpmkv->lock scope */

  /* 部分单列只读操作要求输出格式化到注册在整个 Op 上的、上层另外指明的输出空间 Op::outbl（如果是单独的一块空间的话） */
  // if (op->outbl && result == 0
  //     && op->ops.size() == 1 /*&& op->ops[0].outdata.length()*/
  //     && op->outbl != &op->ops[0].outdata) {
  //   op->outbl->clear();
  //   op->outbl->append(op->ops[0].outdata);
  // }

  return true;
}

}   /* namespace ceph_mock */
